package com.bungarus.model;

import com.bungarus.busi.InnerClient;
import com.bungarus.cluster.ClusterChangeListener;
import com.bungarus.cluster.ClusterManager;
import com.bungarus.cluster.Node;
import com.bungarus.cluster.NodeInfo;
import com.bungarus.config.BungarusSpringConfig;
import com.bungarus.utils.IPAddressUtils;
import com.google.common.collect.Sets;
import com.google.common.net.HostAndPort;
import org.apache.log4j.Logger;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.data.mongodb.core.MongoTemplate;

import java.net.Inet4Address;
import java.net.InetAddress;
import java.security.SecureRandom;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;

/**
 * Created by tang on 2018/12/20.
 */
abstract public class AbstractBungarusServerWithMultipleClients extends AbstractBungarusServer implements ClusterChangeListener{
    final Logger logger = Logger.getLogger(AbstractBungarusServerWithMultipleClients.class);

    private static final ApplicationContext springContext =
            new AnnotationConfigApplicationContext(BungarusSpringConfig.class);

    // multiple inner clients connected
    private final Map<String, List<InnerClient>> mInnerClients = new ConcurrentHashMap<>(0);

    @Override
    public void initClusterAndConnectRemoteNode() throws BungarusServerException{
        try {
            //register myself into a cluster
            register2Cluster();
            ///gather the remote nodes info
            initRemoteNodesInfo();

            for (String CLUSTER_TAG : getRemoteNodeTags()) {
                CopyOnWriteArrayList<NodeInfo> queue = calcRemoteServerNodes(remoteServerNodes.get(CLUSTER_TAG));
                List<InnerClient> innerClients = new CopyOnWriteArrayList<>();
                for(int i=queue.size()-1;i>=0;i--){
                    try {
                        InnerClient innerClient = this.connect(i, queue);
                        innerClients.add(innerClient);
                    }
                    catch (Exception e){
                        e.printStackTrace();
                    }
                }
                // multiple inner clients for each tag
                mInnerClients.put(CLUSTER_TAG, innerClients);
            }
            getClusterManager().addClusterChangeListener(this);
            this.setStatus(Node.STATUS_RUNNING);
        }
        catch (Exception e) {
            throw new BungarusServerException(e);
        }
    }

    /**
     * get just one inner client from multiple connected inner clients for each tag
     * @return
     */
    @Override
    public Map<String, InnerClient> getInnerClients() {
        Map<String, List<InnerClient>> clientsMap = getInnerClientList();
        if(!clientsMap.isEmpty()) {
            Map<String, InnerClient> results = new HashMap<>(0);
            Set<String> keys = clientsMap.keySet();
            keys.forEach(key -> {
                if(!results.containsKey(key)) {
                    if(clientsMap.get(key).size() > 0) {
                        SecureRandom secureRandom = new SecureRandom();
                        InnerClient client;
                        do {
                            long random = secureRandom.nextLong();
                            long index = Math.abs(random == Long.MIN_VALUE ? Long.MAX_VALUE : random) % clientsMap.get(key).size();
                            client = clientsMap.get(key).get((int) index);
                        } while (client.isShuttingDown());
                        results.put(key, client);
                    }
                    else{
                        logger.info("cluster " + key + " has no nodes available.");
                    }
                }
            });
            return results;
        }
        return null;
    }

    @Override
    public MongoTemplate getMongoTemplate4FailedRetry() {
        return (MongoTemplate) springContext.getBean("mongoTemplate");
    }

    public Map<String, List<InnerClient>> getInnerClientList() {
        return mInnerClients;
    }

    @Override
    public List<String> getClusterTags() {
        return Arrays.asList(getRemoteNodeTags());
    }


    @Override
    public void shutdown() throws BungarusServerException {
        mInnerClients.values().forEach(clients -> clients.forEach(InnerClient::close));
        super.shutdown();
    }

    @Override
    public void onClusterNodeRemoved(NodeInfo node) {
    }

    @Override
    public void onClusterNodeUpdated(NodeInfo node) { // A new server is added into the cluster, the mInnerClients must be refreshed.
        IPAddressUtils.HostPortPair hostPortPair = new IPAddressUtils.HostPortPair();
        hostPortPair.host = node.getIp();
        hostPortPair.port = node.getMessagePorts().get(getTag());
        RemoteServerAndInnerClientConnectionMap remoteConnectionsMap = remoteServerAndInnerClientConnectionMap.get(node.getClusterBelongsTo());
        if(!remoteConnectionsMap.hasConnection(hostPortPair)) {
            InnerClient innerClient = createNewInnerClientOnClusterNodeUpdated(node);
            innerClient.connect(0, 2, getTag());

            List<InnerClient> repeat = remoteConnectionsMap.getClientsConnectedToTheSameRemoteServer();
            if(null != repeat) {
                removeClient(node.getClusterBelongsTo(), repeat.get(0));
                repeat.get(0).close();
            }
            addNewClient(node.getClusterBelongsTo(), innerClient);
        }
    }

    protected InnerClient createNewInnerClientOnClusterNodeUpdated(NodeInfo node) {
        return new InnerClient(this, 0, new CopyOnWriteArrayList<>(Collections.singleton(node)), this.getMongoTemplate4FailedRetry());
    }

    private void addNewClient(String remoteServerClusterTag, InnerClient newClient) {
        this.mInnerClients.get(remoteServerClusterTag).add(newClient);
    }

    private void removeClient(String remoteServerClusterTag, InnerClient client) {
        List<InnerClient> oldConnectedClients = this.mInnerClients.get(remoteServerClusterTag);
        oldConnectedClients.remove(client);
    }
}
